package cn.xuexiyuan.flinkstudy.sink;

import cn.xuexiyuan.flinkstudy.entity.Student;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @Description: 自定义 Sink
 * @Author 左龙龙
 * @Date 21-3-24
 * @Version 1.0
 **/
public class SinkDemo02 {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<Student> studentDS = env.fromElements(new Student(null, "flink", 3));

        // 2.transformation

        // 3.sink
        studentDS.addSink(new MySqlSink());

        // 4.excute
        env.execute("SinkDemo02");

    }

}


class MySqlSink extends RichSinkFunction<Student> {

    private Connection connection = null;
    private PreparedStatement ps = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        String url = "jdbc:mysql://localhost:3306/flink_study";
        String user = "root";
        String passwd = "";

        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection(url, user, passwd);
        ps = connection.prepareStatement("insert into tb_student(name, age) values(?,?)");
    }

    @Override
    public void invoke(Student student, Context context) throws Exception {
        ps.setString(1, student.getName());
        ps.setInt(2, student.getAge());

        ps.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        if(null != connection) connection.close();
        if(null != ps) ps.close();
    }

}
